#include "ThreadPool.h"
#include <pthread.h>
#include <semaphore.h>
#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>
#include <stdio.h>
#include <unistd.h>

// 线程池
struct ThreadPool
{
    Task *_taskQueue; // 环形任务队列
    int _capacity; // 环形队列的容量
    int _size; // 环形队列的任务数目
    int _proIndex;        // 生产者下标
    int _conIndex;        // 消费者下标

    pthread_t _administrator; // 管理者线程
    pthread_t *_workers; // 工作者线程数组

    int _minNum; // 工作线程的最小数目
    int _maxNum; // 工作线程的最大数目
    int _busyNum; // 正在执行任务的线程数目
    int _actualNum; // 实际创建的线程数目
    int _exitNum;

    sem_t _proSpace; // 标识生产者还有多少空间资源
    sem_t _conSpace; // 标识消费者还有多少数据资源

    pthread_mutex_t _proLock; // 多生产者之间的互斥锁
    pthread_mutex_t _conLock; // 多消费者之间的互斥锁
    pthread_mutex_t _busyLock;
    pthread_mutex_t _threadPoolLock;
    // pthread_mutex_t _sizeLock;
};

// 工作线程,进行获取,执行任务
void *FetchTask(void *args)
{
    assert(args!=NULL);
    ThreadPool *threadPoolPtr=(ThreadPool *)args;
    while(true)
    {   
        Task task;

        // 对消费者的信号量进行P操作
        sem_wait(&threadPoolPtr->_conSpace);

        pthread_mutex_lock(&threadPoolPtr->_threadPoolLock);
        if(threadPoolPtr->_exitNum>0)
        {
            threadPoolPtr->_exitNum--;
            pthread_mutex_unlock(&threadPoolPtr->_threadPoolLock);
            Thread_Exit(threadPoolPtr);
        }
        pthread_mutex_unlock(&threadPoolPtr->_threadPoolLock);

        pthread_mutex_lock(&threadPoolPtr->_conLock);

        task=threadPoolPtr->_taskQueue[threadPoolPtr->_conIndex++];
        threadPoolPtr->_conIndex%=threadPoolPtr->_capacity;

        pthread_mutex_unlock(&threadPoolPtr->_conLock);
        // 消费者获取完任务后,数据资源减少,空间资源增加,对生产者的信号量进行V操作
        sem_post(&threadPoolPtr->_proSpace);
        
        // pthread_mutex_lock(&threadPoolPtr->_sizeLock);
        // threadPoolPtr->_size--;
        // pthread_mutex_unlock(&threadPoolPtr->_sizeLock);

        pthread_mutex_lock(&threadPoolPtr->_threadPoolLock);
        threadPoolPtr->_size--;
        pthread_mutex_unlock(&threadPoolPtr->_threadPoolLock);

        pthread_mutex_lock(&threadPoolPtr->_busyLock);
        threadPoolPtr->_busyNum++;
        pthread_mutex_unlock(&threadPoolPtr->_busyLock);

        // 执行任务
        task.function(task.args);
        free(task.args);
        sleep(1);

        pthread_mutex_lock(&threadPoolPtr->_busyLock);
        threadPoolPtr->_busyNum--;
        pthread_mutex_unlock(&threadPoolPtr->_busyLock);
    }
}

// 管理线程,根据任务与线程的数量进行调整
void *AdjustThreadNum(void *args)
{
    assert(args!=NULL);
    ThreadPool *threadPoolPtr=(ThreadPool *)args;
    while(true)
    {
        // 每三秒进行一次检测
        sleep(3);
        fprintf(stdout,"管理线程检测\n");

        pthread_mutex_lock(&threadPoolPtr->_threadPoolLock);
        // 如果环形队列的任务数量比实际线程的数量多,而且实际线程数量比最大线程数量小,则进行新增线程
        if(threadPoolPtr->_size > threadPoolPtr->_actualNum && threadPoolPtr->_actualNum<threadPoolPtr->_maxNum)
        {   
            fprintf(stdout,"增加线程\n");
            int count=0;
            int i=0;
            for(i=0;i<threadPoolPtr->_maxNum && count<g_add_thread_num;++i)
            {
                if(threadPoolPtr->_workers[i]==0)
                {
                    pthread_create(threadPoolPtr->_workers+i,NULL,FetchTask,(void *)threadPoolPtr);
                    count++;
                    threadPoolPtr->_actualNum++;
                }
            }
        }
        pthread_mutex_unlock(&threadPoolPtr->_threadPoolLock);

        pthread_mutex_lock(&threadPoolPtr->_threadPoolLock);

        // 如果实际的线程比环形队列中的任务多两倍,而且实际线程数量比最小线程数量大,则进行销毁线程
        if(threadPoolPtr->_actualNum>2*threadPoolPtr->_size&&threadPoolPtr->_actualNum>threadPoolPtr->_minNum)
        {
            fprintf(stdout,"减少线程\n");
            threadPoolPtr->_exitNum=g_sub_thread_num;
            int exitNum=g_sub_thread_num;
            while(exitNum>0 && threadPoolPtr->_actualNum > threadPoolPtr->_minNum)
            {
                sem_post(&threadPoolPtr->_conSpace);
                exitNum--;
                threadPoolPtr->_actualNum--;
            }
            // threadPoolPtr->_exitNum=g_sub_thread_num;
            // while(threadPoolPtr->_exitNum>0 && threadPoolPtr->_actualNum>threadPoolPtr->_minNum)
            // {
            //     pthread_mutex_unlock(&threadPoolPtr->_threadPoolLock);
            //     sem_post(&threadPoolPtr->_conSpace);
            //     pthread_mutex_lock(&threadPoolPtr->_threadPoolLock);
            //     threadPoolPtr->_actualNum--;
            // }
        }
        pthread_mutex_unlock(&threadPoolPtr->_threadPoolLock);
    }
}

void ThreadPool_PushTask(ThreadPool *threadPoolPtr,Task *task)
{
    assert(threadPoolPtr!=NULL && task!=NULL);

    // 对生产者对应的信号量进行P操作
    sem_wait(&threadPoolPtr->_proSpace);
    
    // 多生产者之间的互斥锁
    pthread_mutex_lock(&threadPoolPtr->_proLock);

    threadPoolPtr->_taskQueue[threadPoolPtr->_proIndex++]=*task;
    threadPoolPtr->_proIndex%=threadPoolPtr->_capacity;

    pthread_mutex_unlock(&threadPoolPtr->_proLock);

    pthread_mutex_lock(&threadPoolPtr->_threadPoolLock);
    threadPoolPtr->_size++;
    pthread_mutex_unlock(&threadPoolPtr->_threadPoolLock);

    // 任务推送完成则对消费者对应的信号量进行V操作
    sem_post(&threadPoolPtr->_conSpace);
}

ThreadPool *ThreadPool_Create(const int minNum,const int maxNum,const int capacity)
{
    assert(maxNum>=minNum);
    assert(capacity>0);

    ThreadPool *threadPoolPtr=(ThreadPool *)malloc(sizeof (ThreadPool));
    if(NULL==threadPoolPtr)
    {
        perror("ThreadPool malloc fail");
        return NULL;
    }

    threadPoolPtr->_taskQueue=(Task *)malloc(sizeof (Task)*capacity);
    if(NULL==threadPoolPtr->_taskQueue)
    {
        free(threadPoolPtr);
        perror("threadPoolPtr->_taskQueue malloc fail");
        return NULL;
    }

    threadPoolPtr->_workers=(pthread_t *)malloc(sizeof (pthread_t)*maxNum);
    if(NULL==threadPoolPtr->_workers)
    {
        free(threadPoolPtr->_taskQueue);
        free(threadPoolPtr);
        perror("threadPoolPtr->_workers malloc fail");
        return NULL;
    }

    // 保存最大与最小线程数目
    threadPoolPtr->_maxNum=maxNum; 
    threadPoolPtr->_minNum=minNum;
    // 开始时执行任务的线程为0
    threadPoolPtr->_busyNum=0; 
    // 最开始创建minNum个线程
    threadPoolPtr->_actualNum=minNum; 
    // 
    threadPoolPtr->_exitNum=0;
    // 记录环形任务队列的容量
    threadPoolPtr->_capacity=capacity;
    threadPoolPtr->_proIndex=0;
    threadPoolPtr->_conIndex=0;
    threadPoolPtr->_size=0;

    // 互斥锁的初始化
    pthread_mutex_init(&threadPoolPtr->_conLock,NULL);
    pthread_mutex_init(&threadPoolPtr->_proLock,NULL);
    pthread_mutex_init(&threadPoolPtr->_busyLock,NULL);
    pthread_mutex_init(&threadPoolPtr->_threadPoolLock,NULL);
    // pthread_mutex_init(&threadPoolPtr->_sizeLock,NULL);

    // 消费者与生产者的信号量初始化
    sem_init(&threadPoolPtr->_conSpace,0,0); // 开始时,数据资源为零
    sem_init(&threadPoolPtr->_proSpace,0,capacity); // 空间资源为capacity

    pthread_create(&threadPoolPtr->_administrator,NULL,AdjustThreadNum,(void *)threadPoolPtr);

    // 如果threadPoolPtr->_workers[i]为零,则说明该位置对应的线程无效
    memset(threadPoolPtr->_workers,0,sizeof(pthread_t)*maxNum);

    int i=0;
    for(i=0;i<minNum;++i)
    {
        pthread_create(threadPoolPtr->_workers+i,NULL,FetchTask,threadPoolPtr);
    }
    return threadPoolPtr;
}

void ThreadPool_Destroy(ThreadPool *threadPoolPtr)
{
    assert(threadPoolPtr!=NULL);
    
    // 先将线程池中的所有线程取消

    // 取消管理线程
    pthread_cancel(threadPoolPtr->_administrator);

#ifdef DEBUG
    sleep(1);
#endif

    // 取消工作线程
    int i=0;
    for(i=0;i<threadPoolPtr->_maxNum;++i)
    {
        if(threadPoolPtr->_workers[i]!=0)
            pthread_cancel(threadPoolPtr->_workers[i]);
        #ifdef DEBUG
        sleep(1);
        #endif
    }


    // 释放环形队列,工作线程数组
    free(threadPoolPtr->_taskQueue);
    free(threadPoolPtr->_workers);
    // 最后释放线程池
    free(threadPoolPtr);
}

void Thread_Exit(ThreadPool *threadPoolPtr)
{
    pthread_t tid=pthread_self();
    int i=0;

    pthread_mutex_lock(&threadPoolPtr->_threadPoolLock);
    for(i=0;i<threadPoolPtr->_maxNum;++i)
    {
        if(threadPoolPtr->_workers[i]==tid)
        {
            threadPoolPtr->_workers[i]=0;
        }
    }
    pthread_mutex_unlock(&threadPoolPtr->_threadPoolLock);

    pthread_exit(NULL);
}